package com.example.literj.kt.internal.operators.observable

import com.example.literj.kt.core.Observable
import com.example.literj.kt.core.ObservableEmitter
import com.example.literj.kt.core.ObservableOnSubscribe
import com.example.literj.kt.core.Observer
import com.example.literj.kt.disposables.Disposable
import com.example.literj.kt.internal.disposables.DisposableHelper
import com.example.literj.kt.internal.util.AtomicThrowable
import com.example.literj.kt.internal.util.ExceptionHelper
import com.example.literj.kt.operators.SimpleQueue
import com.example.literj.kt.operators.SpscLinkedArrayQueue
import com.example.literj.kt.plugins.RxJavaPlugins
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicReference

class ObservableCreate<T>(private val mSource: ObservableOnSubscribe<T>) : Observable<T>() {
    override fun subscribeActual(observer: Observer<T>) {
        val parent = CreateEmitter(observer)
        observer.onSubscribe(parent)

        try {
            mSource.subscribe(parent)
        } catch (th: Throwable) {
            parent.onError(th)
        }
    }

    class CreateEmitter<T>(private val mObserver: Observer<T>) : AtomicReference<Disposable>(),
        ObservableEmitter<T>, Disposable {

        override fun onNext(value: T?) {
            if (value == null) {
                onError(ExceptionHelper.createNullPointerException("onNext called with a null value."))
                return
            }
            if (!isDisposed()) {
                mObserver.onNext(value)
            }
        }

        override fun onError(error: Throwable) {
            if (!tryOnError(error)) {
                RxJavaPlugins.onError(error)
            }
        }

        override fun onComplete() {
            if (!isDisposed()) {
                try {
                    mObserver.onComplete()
                } finally {
                    dispose()
                }
            }
        }

        override fun tryOnError(th: Throwable?): Boolean {
            var error = th
            if (error == null) {
                error = ExceptionHelper.createNullPointerException("onError called with a null Throwable.")
            }
            if (!isDisposed()) {
                try {
                    mObserver.onError(error)
                } finally {
                    dispose()
                }
                return true
            }
            return false
        }

        override fun dispose() {
            DisposableHelper.dispose(this)
        }

        override fun isDisposed(): Boolean {
            return DisposableHelper.isDisposed(get())
        }

        override fun setDisposable(d: Disposable) {
            DisposableHelper.set(this, d)
        }

        override fun serialize(): ObservableEmitter<T> {
            return SerializedEmitter(this)
        }
    }

    class SerializedEmitter<T>(
        private val emitter: ObservableEmitter<T>,
        private val errors: AtomicThrowable = AtomicThrowable(),
        private val queue: SpscLinkedArrayQueue<T> = SpscLinkedArrayQueue(16)
    ) : AtomicInteger(), ObservableEmitter<T> {
        @Volatile
        private var done = false

        override fun onNext(value: T?) {
            if (done || emitter.isDisposed()) {
                return
            }
            if (value == null) {
                onError(ExceptionHelper.createNullPointerException("onNext called with a null value."))
                return
            }
            if (get() == 0 && compareAndSet(0, 1)) {
                emitter.onNext(value)
                if (decrementAndGet() == 0) {
                    return
                }
            } else {
                val q: SimpleQueue<T> = queue
                synchronized(q) { q.offer(value) }
                if (andIncrement != 0) {
                    return
                }
            }
            drainLoop()
        }

        override fun onError(error: Throwable) {
            if (!tryOnError(error)){
                RxJavaPlugins.onError(error)
            }
        }

        override fun onComplete() {
            if (done || emitter.isDisposed()) {
                return
            }
            done = true
            drain()
        }

        override fun setDisposable(d: Disposable) {
            emitter.setDisposable(d)
        }

        override fun isDisposed(): Boolean {
            return emitter.isDisposed()
        }

        override fun serialize(): ObservableEmitter<T> {
            return this
        }

        override fun tryOnError(th: Throwable?): Boolean {
            if (done || emitter.isDisposed()) {
                return false
            }
            var t = th
            if (t == null) {
                t = ExceptionHelper.createNullPointerException("onError called with a null Throwable.")
            }
            if (errors.tryAddThrowable(t)) {
                done = true
                drain()
                return true
            }
            return false
        }

        private fun drain() {
            if (andIncrement == 0) {
                drainLoop()
            }
        }

        private fun drainLoop() {
            val e = emitter
            val q = queue
            val errors: AtomicThrowable = this.errors
            var missed = 1
            while (true) {
                while (true) {
                    if (e.isDisposed()) {
                        q.clear()
                        return
                    }
                    if (errors.get() != null) {
                        q.clear()
                        errors.tryTerminateConsumer(e)
                        return
                    }
                    val d = done
                    val v = q.poll()
                    val empty = v == null
                    if (d && empty) {
                        e.onComplete()
                        return
                    }
                    if (empty) {
                        break
                    }
                    e.onNext(v)
                }
                missed = addAndGet(-missed)
                if (missed == 0) {
                    break
                }
            }
        }

        override fun toByte(): Byte {
            return get().toByte()
        }

        override fun toChar(): Char {
            return get().toChar()
        }

        override fun toShort(): Short {
            return get().toShort()
        }
    }
}